#include "ul_condition_task_queue.h"
#include <stdlib.h>
#include <string.h>
#include "ul_define.h"
#include "ul_malloc.h"
using namespace LinuxNetworkUtility;
namespace 
{
  static const uint64_t CONDITION_TASK_MAGIC = 0xc6e7965f2c429546; 
}

LinuxNetworkUtility::UlConditionQueenWorkerTQ::UlConditionQueenWorkerTQ()
{
  worker_num_ = 0;
  cond_pipes_ = NULL;
  next_push_pipe_idx_ = 0;
}

void LinuxNetworkUtility::UlConditionQueenWorkerTQ::clear()
{
  DLink *it = NULL;
  UlTsk *tsk = NULL;
  if (NULL != cond_pipes_)
  {
    for (int32_t i = 0; i < worker_num_; i++)
    {
      pthread_cond_destroy(&(cond_pipes_[i].cond_));
      pthread_mutex_destroy(&(cond_pipes_[i].mutex_));
      pthread_spin_destroy(&(cond_pipes_[i].spin_));
      cond_pipes_[i].pending_tsk_num_ = 0;
      it = cond_pipes_[i].worker_end_list_.next();
      while (it != &(cond_pipes_[i].worker_end_list_))
      {
        tsk = CONTAINING_RECORD(it, UlTsk, link_);
        it = it->next();
        UL_LOG_WARNING("task not processed by worker bee [ptr:%p]",tsk);
        tsk->link_.remove();
        ul_free(tsk);
        tsk = NULL;
      }
      it = cond_pipes_[i].queen_end_list_.next();
      while (it != &(cond_pipes_[i].queen_end_list_))
      {
        tsk = CONTAINING_RECORD(it, UlTsk, link_);
        it = it->next();
        UL_LOG_WARNING("response not processed by queen bee [ptr:%p]",tsk);
        tsk->link_.remove();
        ul_free(tsk);
        tsk = NULL;
      }
    }
    delete [] cond_pipes_;
    cond_pipes_ = NULL;
  }
  worker_num_ = 0;
  next_push_pipe_idx_ = 0;
}

LinuxNetworkUtility::UlConditionQueenWorkerTQ::~UlConditionQueenWorkerTQ()
{
  clear();
}

int LinuxNetworkUtility::UlConditionQueenWorkerTQ::init(const int64_t worker_bee_num)
{
  int err = 0;
  if (NULL != cond_pipes_)
  {
    UL_LOG_WARNING("task queue was already initialized");
    errno = EALREADY;
    err =  ERRNO2ERR(errno);
  }
  if (worker_bee_num <= 0)
  {
    UL_LOG_WARNING("invalid argument [worker_bee_num:%lld]", worker_bee_num);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }

  if (0 == err)
  {
    worker_num_ = worker_bee_num;
    cond_pipes_ = new(std::nothrow) UlConditionPipe[worker_bee_num];
    if (NULL == cond_pipes_)
    {
      UL_LOG_WARNING("fail to allocate memory for fd pair array [errno:%d]", errno);
      err = ERRNO2ERR(errno);
    }
  }
  if ((NULL != cond_pipes_) && 0 != err)
  {
    clear();
  }
  return err;
}

void* LinuxNetworkUtility::UlConditionQueenWorkerTQ::malloc_tsk(const int64_t size, const int32_t mod_id)
{
  UlTsk *ltsk = NULL;
  void *ptr = NULL;
  void *result = NULL;
  if (size > 0)
  {
    ptr = ul_malloc(size + sizeof(UlTsk),mod_id,NULL,FREE_TO_ALLOCATED_THREAD_POOL);
    if (ptr != NULL)
    {
      ltsk = new(ptr)UlTsk();
      ltsk->magic_ = CONDITION_TASK_MAGIC;
      result = ltsk->tsk_;
    }
  }
  return result;
}

void LinuxNetworkUtility::UlConditionQueenWorkerTQ::free_tsk(void* tsk)
{
  int err = 0;
  UlTsk *l_tsk  = NULL;
  if ( 0 == err && NULL == tsk)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("argument error [tsk:%p]", tsk);
  }
  else
  {
    l_tsk = reinterpret_cast<UlTsk*>((char*)tsk - sizeof(UlTsk));
  }
  if (0 == err && CONDITION_TASK_MAGIC != l_tsk->magic_)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_ERROR("task magic error [tsk:%p,magic:%llu,expect_magic:%llu]", l_tsk, 
                 l_tsk->magic_, CONDITION_TASK_MAGIC);
  }
  if (0 == err)
  {
    ul_free(l_tsk);
  }
}

int LinuxNetworkUtility::UlConditionQueenWorkerTQ::dispatch_task(void *tsk)
{
  int err = 0;
  if (NULL == cond_pipes_)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("task queue not initialized yet, initialize it first");
  }
  UlTsk *l_tsk = NULL;
  int64_t cur_pipe_idx = next_push_pipe_idx_;
  if ( 0 == err &&NULL == tsk)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("argument error [tsk:%p]", tsk);
  }
  else
  {
    l_tsk = reinterpret_cast<UlTsk*>((char*)tsk - sizeof(UlTsk));
  }
  if (0 == err && CONDITION_TASK_MAGIC != l_tsk->magic_)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_ERROR("task magic error [tsk:%p,magic:%llu,expect_magic:%llu]", l_tsk, 
                 l_tsk->magic_, CONDITION_TASK_MAGIC);
  }
  if (0 == err)
  {
    /// find first idle consumer
    int64_t checked_fd_num = 0;
    for (checked_fd_num = 0; checked_fd_num < worker_num_; checked_fd_num++)
    {
      if (cond_pipes_[cur_pipe_idx].pending_tsk_num_ == 0)
      {
        cond_pipes_[cur_pipe_idx].enqueue_task(l_tsk);
        next_push_pipe_idx_ = (cur_pipe_idx + 1)%worker_num_;
        break;
      }
      else
      {
        cur_pipe_idx ++;
        cur_pipe_idx = cur_pipe_idx%worker_num_;
      }
    }
    /// there is no idle consumer, find first correct consumer
    cur_pipe_idx = next_push_pipe_idx_;
    if (checked_fd_num >= worker_num_)
    {
      cur_pipe_idx = next_push_pipe_idx_;
      cond_pipes_[cur_pipe_idx].enqueue_task(l_tsk);
      next_push_pipe_idx_ = (cur_pipe_idx + 1)%worker_num_;
    }
    else
    {
      /// do nothing, find one idle pipe;
    }
  }
  return err;
}

int LinuxNetworkUtility::UlConditionQueenWorkerTQ::claim_task(void **tsk, const int64_t ntsk,  
                                                              const int64_t timeout_us, 
                                                              const int64_t worker_id)
{
  int err = 0;
  UNUSED(timeout_us);
  if (NULL == tsk || ntsk <= 0)
  {
    UL_LOG_WARNING("argument error [tsk:%p,ntsk:%lld]", tsk, ntsk);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (0 == err && (worker_id < 0 || worker_id >= worker_num_))
  {
    UL_LOG_WARNING("argument error, consumer id out of range [worker_i:%lld,consumer_num:%lld]", 
                   worker_id, worker_num_);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (0 == err && NULL == cond_pipes_)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("task queue not initialized yet, initialize it first");
  }
  if (0 == err)
  {
    err = cond_pipes_[worker_id].claim_task(tsk,ntsk, timeout_us);
  }
  return err;
}

int LinuxNetworkUtility::UlConditionQueenWorkerTQ::submit_output(void *tsk, const int64_t worker_id)
{
  int err = 0;
  UlTsk *l_tsk = NULL;
  if (NULL == tsk)
  {
    UL_LOG_WARNING("argument error [tsk:%p]", tsk);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (0 == err && (worker_id < 0 || worker_id >= worker_num_))
  {
    UL_LOG_WARNING("argument error, consumer id out of range [worker_i:%lld,consumer_num:%lld]", 
                   worker_id, worker_num_);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if ( 0 == err &&NULL == tsk)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_WARNING("argument error [tsk:%p]", tsk);
  }
  else
  {
    l_tsk = reinterpret_cast<UlTsk*>((char*)tsk - sizeof(UlTsk));
  }
  if (0 == err && CONDITION_TASK_MAGIC != l_tsk->magic_)
  {
    errno = EINVAL;
    err = ERRNO2ERR(errno);
    UL_LOG_ERROR("task magic error [tsk:%p,magic:%llu,expect_magic:%llu]", l_tsk, 
                 l_tsk->magic_, CONDITION_TASK_MAGIC);
  }
  if (0 == err)
  {
    cond_pipes_[worker_id].submit_output(l_tsk);
  }
  return err;
}

int LinuxNetworkUtility::UlConditionQueenWorkerTQ::fetch_output(void **tsk, const int64_t ntsk, 
                                                                const int64_t timeout_us)
{
  int err = 0;
  int64_t poll_timeout_ms = 0;
  int64_t got_tsk_num = 0;
  if (NULL == tsk || ntsk <= 0)
  {
    UL_LOG_WARNING("param error [tsk:%p,ntsk:%lld]", tsk, ntsk);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (timeout_us <= 0)
  {
    poll_timeout_ms = 0;
  }
  else
  {
    poll_timeout_ms = (timeout_us + 999)/1000;
  }
  if (0 == err)
  {
    for (int32_t i = 0; i < worker_num_ && got_tsk_num < ntsk; i++)
    {
      int64_t cur_got_num  = 0;
      cur_got_num = cond_pipes_[i].fetch_output(tsk + got_tsk_num,ntsk - got_tsk_num);
      got_tsk_num += cur_got_num;
    }
  }
  if (0 == err)
  {
    err = got_tsk_num;
  }
  return err;
}
